package com.chb.storm.bolt;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.storm.hbase.bolt.AbstractHBaseBolt;
import org.apache.storm.hbase.bolt.HBaseBolt;
import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
import org.apache.storm.hbase.common.HBaseClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.Config;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;

public class AdHBaseBolt extends HBaseBolt {

	 private static final Logger LOG = LoggerFactory.getLogger(AdHBaseBolt.class);
	
	public AdHBaseBolt(String tableName, HBaseMapper mapper) {
		super(tableName, mapper);
		// TODO Auto-generated constructor stub
	}

	@Override
	public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
        this.collector = collector;
        final Configuration hbConfig = HBaseConfiguration.create();

        Map<String, Object> hbaseConfMap;
        if (StringUtils.isBlank(configKey))
        {
        	hbaseConfMap = new HashMap<String, Object>();
        }
        else
        {
        	Map<String, Object> conf = (Map<String, Object>)map.get(this.configKey);
        	
        	if (null != conf)
        	{
        		if(conf.get("hbase.rootdir") == null) {
        			LOG.warn("No 'hbase.rootdir' value found in configuration! Using HBase defaults.");
        		}
        		for(String key : conf.keySet()) {
        			hbConfig.set(key, String.valueOf(conf.get(key)));
        		}
        		
        		hbaseConfMap = new HashMap<String, Object>(conf);
        	}
        	else
        	{
        		hbaseConfMap = new HashMap<String, Object>();
			}
		}
        

        //heck for backward compatibility, we need to pass TOPOLOGY_AUTO_CREDENTIALS to hbase conf
        //the conf instance is instance of persistentMap so making a copy.
//        Map<String, Object> hbaseConfMap = new HashMap<String, Object>(conf);
        
        hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, map.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
        this.hBaseClient = new HBaseClient(hbaseConfMap, hbConfig, tableName);
    
	}

	
}
